package com.easypay.reactive8583.client;

import com.easypay.reactive8583.AbstractIso8583Connector;
import com.easypay.reactive8583.iso.MessageFactory;
import com.easypay.reactive8583.netty.pipeline.CompositeIsoMessageHandler;
import com.easypay.reactive8583.netty.pipeline.Iso8583ChannelInitializer;
import com.easypay.reactive8583.netty.pipeline.ReconnectOnCloseListener;
import com.solab.iso8583.IsoMessage;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.AllArgsConstructor;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.concurrent.TimeUnit;


/**
 * @author pandans
 * @date 2021/03/04 17:55
 * @Description:
 */
public class Iso8583Client<T extends IsoMessage> extends AbstractIso8583Connector<ClientConfiguration, Bootstrap, T> {
    private SocketAddress socketAddress;
    private ClientConfiguration config;
    private MessageFactory<T> isoMessageFactory;
    private ReconnectOnCloseListener reconnectOnCloseListener;


    public Iso8583Client(SocketAddress socketAddress, ClientConfiguration configuration, MessageFactory<T> isoMessageFactory) {
        super(configuration, isoMessageFactory);
        this.socketAddress = socketAddress;
        this.config = configuration;
        this.isoMessageFactory = isoMessageFactory;
    }

    public Iso8583Client(ClientConfiguration configuration, MessageFactory<T> isoMessageFactory, CompositeIsoMessageHandler<T> messageHandler) {
        super(configuration, isoMessageFactory, messageHandler);
    }

    @Override
    protected Bootstrap createBootstrap() {
        Bootstrap b = new Bootstrap();
        b.group(bossEventLoopGroup).channel(NioSocketChannel.class)
            .remoteAddress(socketAddress)
                .handler(
                        new Iso8583ChannelInitializer<Channel, Bootstrap, ClientConfiguration>(
                                configuration,
                        configurer,
                        workerEventLoopGroup,
                                (MessageFactory<IsoMessage>)isoMessageFactory,
                        getMessageHandler()
                )
            );
        configureBootstrap(b);
        b.validate();
        reconnectOnCloseListener = new ReconnectOnCloseListener(
                this,
                configuration.getReconnectInterval(),
                bossEventLoopGroup
        );
        return b;
    }
    public ChannelFuture connect(SocketAddress serverAddress) throws InterruptedException {
        this.socketAddress = serverAddress;
        return connect().sync();
    }
    public ChannelFuture connect() throws InterruptedException {
        Channel channel = connectAsync().sync().channel();
        if(channel == null) {
            throw new IllegalStateException("Channel must be set");
        }
        return channel.closeFuture();
    }

    public ChannelFuture connect(String host, int port) throws InterruptedException  {
        return connect(new InetSocketAddress(host, port));
    }

    /**
     * Connects asynchronously to remote address.
     *
     * @return Returns the [ChannelFuture] which will be notified when this
     * channel is active.
     */
    public ChannelFuture connectAsync() {
        logger.debug("Connecting to {}", socketAddress);
        Bootstrap b = bootstrap;
        reconnectOnCloseListener.requestReconnect();
        ChannelFuture connectFuture = b.connect();
        connectFuture.addListener(var1->{
            if (!connectFuture.isSuccess()) {
                reconnectOnCloseListener.scheduleReconnect();
            }
            Channel channel = connectFuture.channel();
            if(channel != null) {
                logger.debug("Client is connected to {}", channel.remoteAddress());
                channel.closeFuture().addListener(reconnectOnCloseListener);
                setChannel(channel);
                return;
            }
        });
        return connectFuture;
    }

    public ChannelFuture disconnectAsync() {
        reconnectOnCloseListener.requestDisconnect();
        Channel channel = this.getChannel();
        logger.info("Closing connection to {}", socketAddress);
        if(channel == null) {
            return null;
        }
        return channel.close();
    }

    public void disconnect() throws InterruptedException{
        disconnectAsync().await();
    }

    /**
     * Sends asynchronously and returns a [ChannelFuture]
     *
     * @param isoMessage A message to send
     * @return ChannelFuture which will be notified when message is sent
     */
    public ChannelFuture sendAsync(IsoMessage isoMessage) {
        Channel ch = getChannel();
        if(ch != null && ch.isWritable()) {
        }else {
            throw new IllegalStateException("Channel is not writable");
        }
        return ch.writeAndFlush(isoMessage);
    }

    /**
     * Sends message synchronously
     */
    public void send(IsoMessage isoMessage) throws InterruptedException{
        sendAsync(isoMessage).sync().await();
    }

    /**
     * Sends message synchronously with timeout
     */
    public void send(IsoMessage isoMessage, Long timeout, TimeUnit timeUnit) throws InterruptedException{
        sendAsync(isoMessage).sync().await(timeout, timeUnit);
    }

    public boolean isConnected() {
        Channel channel = getChannel();
        return channel != null && channel.isActive();
    }
}
